package com.springstack.kafka.consumer.Listener;


import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author qiangfanghao
 * @date 2022/7/25
 * kafka 消费者监听
 *
 */
@Slf4j
@Component
public class KafkaConsumerListener {

    /**
     * 使用config中的配置信息
     * @param record
     */
    @KafkaListener(topics = "test")
    public void listener(ConsumerRecord<String,String> record){
        //获取消息
        String message = record.value();
        //消息偏移量
        long offset = record.offset();
        log.debug("listener读取的消息:{},当前偏移量{}",message,offset);
    }

    /**
     * 使用注解配置信息
     * @param record
     */
    @KafkaListener(topics = "test",groupId = "plane2")
    public void listener2(ConsumerRecord<String,String> record){
        //获取消息
        String message = record.value();
        //消息偏移量
        long offset = record.offset();
        log.debug("listener2读取的消息:{},当前偏移量{}",message,offset);
    }

    /**
     * 使用注解指定 listenerContainerFactory
     * 注意 listenerContainerFactory 开启了批量处理消息
     * @param List<ConsumerRecord<String,String>>
     */
    @KafkaListener(topics = "test",containerFactory = "listenerContainerFactory")
    public void listener3(List<ConsumerRecord<String,String>> recordList){
        ConsumerRecord<String, String> record = recordList.get(0);
        //获取消息
        String message = record.value();
        //消息偏移量
        long offset = record.offset();
        log.debug("listener3读取的消息:{},当前偏移量{}",message,offset);
    }

}
